package com.rstreams;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import netscape.javascript.JSObject;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.streams.core.RocketMQStream;
import org.apache.rocketmq.streams.core.function.ValueMapperAction;
import org.apache.rocketmq.streams.core.rstream.StreamBuilder;
import org.apache.rocketmq.streams.core.serialization.KeyValueSerializer;
import org.apache.rocketmq.streams.core.topology.TopologyBuilder;
import org.apache.rocketmq.streams.core.util.Pair;
import org.apache.rocketmq.streams.core.window.Time;
import org.apache.rocketmq.streams.core.window.WindowInfo;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author: LCG
 * @date: 2023-01-30 14:08:58
 * @description:  简单的测试数据统计
 **/
public class RocketStreamsApp {
    public static void main(String[] args) {

        /* <dependency>   使用该包的代码
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-streams</artifactId>
            <version>1.1.0</version>
        </dependency>*/
        StreamBuilder streamBuilder=new StreamBuilder("wordCount02");
        streamBuilder.source("TEST-12", total -> {
            String value = new String(total, StandardCharsets.UTF_8);
            System.out.println("source----->"+value);
            return new Pair<>(null, value);
        }).filter(msg->{
            if(msg==null||"".equals(msg)||msg.toString().contains("LCG")){
                return false;
            }else{
                return true;
            }
        }).flatMap(msg->{
            //解析处理数据
            String[] s = msg.split(" ");
            return Arrays.asList(s);
        }).keyBy(value->value)
                .count()
                //设置数据输出的位置
                .sink("TEST-15", new KeyValueSerializer<String, Integer>() {
                    @Override
                    public byte[] serialize(String s, Integer integer) throws Throwable {
                        System.out.println("s--->"+s+"----integer--->"+integer);
                        JSONObject jsonObject=new JSONObject();
                        jsonObject.put("wordName",s);
                        jsonObject.put("wordCount",integer);
                        byte[] bytes = jsonObject.toJSONString().getBytes();
                        return bytes;
                    }
                });
        TopologyBuilder topologyBuilder = streamBuilder.build();
        Properties properties = new Properties();
        properties.put(MixAll.NAMESRV_ADDR_PROPERTY, "192.168.137.12:9876");

        RocketMQStream rocketMQStream = new RocketMQStream(topologyBuilder, properties);

        rocketMQStream.start();

        /*DataStreamSource dataStreamSource = StreamBuilder.dataStream("NameSpace", "Job1");

        DataStream dataStream = dataStreamSource.fromRocketmq("TEST-12", "TEST-123", true, "192.168.137.12:9876");
        DataStream dataStream1 = dataStream.map(msg -> {
            return null;
        }).window(TumblingWindow.of(Time.seconds(10), "creteTime"))
                .groupBy("userName")
                .avg("age", "avageAge")
                .count("myCount")
                .setLocalStorageOnly(false)
                .toDataSteam();*/

    }
}
